/*
 * Copyright (C) 2008 CERNET Network Center 
 * dvping/dvmcast/hdvping/hdvmcast series 
 * 
 * Design and coding: 
 *  	Xing Li <xing@cernet.edu.cn> 
 *	Congxiao Bao <congxiao@cernet.edu.cn>
 * 	Jinpeng Jiang <jjp02@mails.tsinghua.edu.cn>
 * 
 * Contributions:
 * 
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 */


#include "dvsource.h"
#include "dvdest.h"

IDVSource::IDVSource()
{
	thread = NullThread;
	quitThread = false;
}
IDVSource::~IDVSource()
{
	Stop();
	for(u_int32_t i = 0; i < dests.size(); ++i)
	{
		delete dests[i];
	}
}
bool IDVSource::AddDestination(IDVDestination* pDest)
{
	if(pDest == NULL)
		return false;
	pDest->SetDVSource(this);
	dests.push_back(pDest);
	return true;
}
bool IDVSource::Start()
{
	quitThread = false;
	thread = NullThread;
	return Thread::Create(thread, ThreadDelegate(this, &IDVSource::SourceProc));
}
bool IDVSource::Stop()
{
	if(thread != NullThread)
	{
		quitThread = true;
		evtThreadQuit.WaitForSignal(5000000);
		if(thread != NullThread)
		{
			Thread::Kill(thread);
			thread = NullThread;
		}
	}
	return true;
}

void* IDVSource::SourceProc(void*)
{
	if(!Init())
		return NULL;
	traffic = 30.0 / 8.0;
	pktQueue.Init();
	
	for(u_int32_t i = 0; i < dests.size(); ++i)
	{
		dests[i]->Start();
	}
	IPEndPoint ep;
	int64_t bytes_read = 0;
	int64_t start_time = DateTime::Now().GetTickCount();

	DVPacket pkt;
	while(!quitThread)
	{
		pkt = GetPacket();
		if(pkt.pdata != NULL)
		{
			bytes_read += pkt.length;
			if(bytes_read > 5000000)
			{
				int64_t now = DateTime::Now().GetTickCount();
				traffic = (double)(bytes_read) / (double)(now - start_time);
				//cout << "traffic:" << 8000000.0 * traffic << endl;
				start_time = now;
				bytes_read = 0;
			}
			pktQueue.AddPacket(pkt);
		}
		else
		{
			Thread::Sleep(100);
		}
	}
	for(u_int32_t i = 0; i < dests.size(); ++i)
	{
		dests[i]->Stop();
	}
	pktQueue.Cleanup();
	Cleanup();
	thread = NullThread;
	evtThreadQuit.Signal();
	return NULL;
}
bool IDVSource::Init()
{
	// empty implementation
	return true;
}
bool IDVSource::Cleanup()
{
	// empty implementation
	return true;
}

packet_iterator IDVSource::GetFirstPacket()
{
	return pktQueue.GetFirstPacket();
}
char* IDVSource::GetNextPacket(packet_iterator& it, u_int32_t& length, IPEndPoint& ep, u_int32_t& pkt_left)
{
	return pktQueue.GetNextPacket(it, length, ep, pkt_left);
}
bool IDVSource::ReleaseIterator(packet_iterator it)
{
	return pktQueue.ReleaseIterator(it);
}
u_int32_t IDVSource::GetPacketCount()
{
	return pktQueue.GetPacketCount();
}
double IDVSource::GetTraffic()
{
	return traffic;
}
bool IDVSource::IsRunning()
{
	return thread != NullThread;
}

